package p.ithorns.framework.common.future;

import p.ithorns.framework.common.reflect.UnsafeUtil;

import javax.validation.constraints.NotNull;
import java.util.concurrent.*;
import java.util.concurrent.locks.LockSupport;

/**
 * SyncFuture
 * <p>
 * 基于FutureTask改造, 使用外部线程任务回调返回同步数据
 * (取除RunnableFuture部分)
 *
 * @author ithorns [2021/12/11 20:07]
 * @version 1.0
 */
public class SyncFuture<V> implements Future<V> {

    private final int state;
    private static final int NEW = 0;
    private static final int COMPLETING = 1;
    private static final int NORMAL = 2;
    private static final int EXCEPTIONAL = 3;
    private static final int CANCELLED = 4;
    private static final int INTERRUPTING = 5;
    private static final int INTERRUPTED = 6;

    private static final long WAITERS_OFFSET = UnsafeUtil.objectFieldOffset(SyncFuture.class, "waiters");

    private static final long STATE_OFFSET = UnsafeUtil.objectFieldOffset(SyncFuture.class, "state");

    /**
     * The result to return or exception to throw from get()
     * non-volatile, protected by state reads/writes
     */
    private Object outcome;

    /**
     * Treiber stack of waiting threads
     */
    private volatile SyncFuture.WaitNode waiters;

    /**
     * Returns result or throws exception for completed task.
     *
     * @param s completed state value
     */
    @SuppressWarnings("unchecked")
    private V report(int s) throws ExecutionException {
        Object x = outcome;
        if (s == NORMAL) {
            return (V) x;
        }
        if (s >= CANCELLED) {
            throw new CancellationException();
        }
        throw new ExecutionException((Throwable) x);
    }

    /**
     * Creates a {@code SyncFuture} that will, upon running, execute the
     * given {@code Callable}.
     *
     * @throws NullPointerException if the callable is null
     */
    public SyncFuture() {
        // ensure visibility of callable
        this.state = NEW;
    }

    @Override
    public boolean isCancelled() {
        return state >= CANCELLED;
    }

    @Override
    public boolean isDone() {
        return state != NEW;
    }

    /**
     * Sets the result of this future to the given value unless
     * this future has already been set or has been cancelled.
     *
     * @param v the value
     */
    protected void set(V v) {
        if (UnsafeUtil.compareAndSwap(this, STATE_OFFSET, NEW, COMPLETING)) {
            outcome = v;
            // final state
            UnsafeUtil.putOrderedInt(this, STATE_OFFSET, NORMAL);
            finishCompletion();
        }
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        if (!(state == NEW &&
                UnsafeUtil.compareAndSwap(this, STATE_OFFSET, NEW,
                        mayInterruptIfRunning ? INTERRUPTING : CANCELLED))) {
            return false;
        }
        try {    // in case call to interrupt throws exception
            if (mayInterruptIfRunning) {
                UnsafeUtil.putOrderedInt(this, STATE_OFFSET, INTERRUPTED);
            }
        } finally {
            finishCompletion();
        }
        return true;
    }

    /**
     * @throws CancellationException - {@inheritDoc}
     */
    @Override
    public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING) {
            s = awaitDone(false, 0L);
        }
        return report(s);
    }

    /**
     * @throws CancellationException -{@inheritDoc}
     */
    @Override
    public V get(long timeout, @NotNull TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException {
        if (unit == null) {
            throw new NullPointerException();
        }
        int s = state;
        if (s <= COMPLETING && (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING) {
            throw new TimeoutException();
        }
        return report(s);
    }

    /**
     * Protected method invoked when this task transitions to state
     * {@code isDone} (whether normally or via cancellation). The
     * default implementation does nothing.  Subclasses may override
     * this method to invoke completion callbacks or perform
     * bookkeeping. Note that you can query status inside the
     * implementation of this method to determine whether this task
     * has been cancelled.
     */
    protected void done() {
    }

    public void setWaiters(WaitNode waiters) {
        this.waiters = waiters;
    }


    /**
     * Simple linked list nodes to record waiting threads in a Treiber
     * stack.  See other classes such as Phaser and SynchronousQueue
     * for more detailed explanation.
     */
    static final class WaitNode {
        volatile Thread thread;
        volatile SyncFuture.WaitNode next;

        WaitNode() {
            thread = Thread.currentThread();
        }
    }

    /**
     * Removes and signals all waiting threads, invokes done(), and
     * nulls out callable.
     */
    private void finishCompletion() {
        // assert state > COMPLETING;
        for (SyncFuture.WaitNode q; (q = waiters) != null; ) {
            if (UnsafeUtil.compareAndSwapObject(this, WAITERS_OFFSET, q, null)) {
                for (; ; ) {
                    Thread t = q.thread;
                    if (t != null) {
                        q.thread = null;
                        LockSupport.unpark(t);
                    }
                    SyncFuture.WaitNode next = q.next;
                    if (next == null) {
                        break;
                    }
                    // unlink to help gc
                    q.next = null;
                    q = next;
                }
                break;
            }
        }

        done();
    }

    /**
     * Awaits completion or aborts on interrupt or timeout.
     *
     * @param timed true if use timed waits
     * @param nanos time to wait, if timed
     * @return state upon completion
     */
    private int awaitDone(boolean timed, long nanos)
            throws InterruptedException {
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
        SyncFuture.WaitNode q = null;
        boolean queued = false;
        for (; ; ) {
            if (Thread.interrupted()) {
                removeWaiter(q);
                throw new InterruptedException();
            }

            int s = state;
            if (s > COMPLETING) {
                return s;
            }
            // cannot time out yet
            else if (s == COMPLETING) {
                Thread.yield();
            } else if (q == null) {
                q = new WaitNode();
            } else if (!queued) {
                queued = UnsafeUtil.compareAndSwapObject(this, WAITERS_OFFSET, q.next = waiters, q);
            } else if (timed) {
                nanos = deadline - System.nanoTime();
                if (nanos <= 0L) {
                    removeWaiter(q);
                    return state;
                }
                LockSupport.parkNanos(this, nanos);
            } else {
                LockSupport.park(this);
            }
        }
    }

    /**
     * Tries to unlink a timed-out or interrupted wait node to avoid
     * accumulating garbage.  Internal nodes are simply unspliced
     * without CAS since it is harmless if they are traversed anyway
     * by releasers.  To avoid effects of unsplicing from already
     * removed nodes, the list is retraversed in case of an apparent
     * race.  This is slow when there are a lot of nodes, but we don't
     * expect lists to be long enough to outweigh higher-overhead
     * schemes.
     */
    private void removeWaiter(SyncFuture.WaitNode node) {
        if (node != null) {
            node.thread = null;
            retry:
            // restart on removeWaiter race
            for (; ; ) {
                for (SyncFuture.WaitNode pred = null, q = waiters, s; q != null; q = s) {
                    s = q.next;
                    if (q.thread != null) {
                        pred = q;
                    } else if (pred != null) {
                        pred.next = s;
                        // check for race
                        if (pred.thread == null) {
                            continue retry;
                        }
                    } else if (!UnsafeUtil.compareAndSwapObject(this, WAITERS_OFFSET, q, s)) {
                        continue retry;
                    }
                }
                break;
            }
        }
    }


    /**
     * Causes this future to report an {@link ExecutionException}
     * with the given throwable as its cause, unless this future has
     * already been set or has been cancelled.
     *
     * @param t the cause of failure
     */
    private void setException(Throwable t) {
        if (UnsafeUtil.compareAndSwap(this, STATE_OFFSET, NEW, COMPLETING)) {
            outcome = t;
            // final state
            UnsafeUtil.putOrderedInt(this, STATE_OFFSET, EXCEPTIONAL);
            finishCompletion();
        }
    }

}
